package com.kingsoft.dc.khaos.module.spark.preprocess.specific.quality

import com.kingsoft.dc.khaos.module.spark.metadata.preprocess.specific.quality.OperatorFilter
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.catalyst.encoders.RowEncoder
import org.apache.spark.sql.{DataFrame, Row}

import scala.collection.mutable.ArrayBuffer

/**
  * 数据格式检查
  *
  *
  */
class DataFormatCheck extends Logging with Serializable {
  def analysisData(dataFrame: DataFrame, operatorFilter: OperatorFilter, field: String): DataFrame = {
    val encoder = RowEncoder(dataFrame.schema)
    val condition = operatorFilter.condition.get
    val expressions: List[String] = operatorFilter.expressions
    var errorData: DataFrame = null
    try {
      errorData = dataFrame.filter(line => line.getAs(field) != null).flatMap(line => {
        val value: String = line.getAs(field).toString
        val rows = ArrayBuffer[Row]()
        var records = 0
        expressions.foreach(ex => {
          val str = ex.split("-<>-")
          val strings1 = str(1).split(" ")
          val str0 = strings1(0).toInt //起始位
          var first = str0 - 1
          var str2 = 0
          if (strings1(2).equals("#")) {
            str2 = value.length //终止位
          } else {
            str2 = strings1(2).toInt
          }
          val str1: String = strings1(1)
          if (str2 > value.length) {
            str2 = value.length
          }
          if (first <= str2) {
            //如果处理后起始位仍然大于终止位则跳过这行处理
            val matchValue = value.substring(first, str2)
            var pattern = ""
            str1 match {
              case "anyNumber" => {
                //所有数字
                pattern = "^[0-9]+$"
              }
              case "anyEnglishLetter" => {
                //所有字母
                pattern = "^[a-zA-Z]+$"
              }
              case "startWithNumber" => {
                //匹配数字开始
                pattern = "^[0-9][a-zA-Z0-9_]+$"
              }
              case "startWithLetter" => {
                //匹配字母开始
                pattern = "^[a-zA-Z][a-zA-Z0-9_]+$"
              }
              case "endWithNumber" => {
                //匹配数字结束
                pattern = "^[a-zA-Z0-9_]+[0-9]$"
              }
              case "endWithLetter" => {
                //匹配字母结束
                pattern = "^[a-zA-Z0-9_]+[a-zA-Z]$"
              }
              case "endWithLetterOrNumber" => {
                //字母或数字结束
                pattern = "^[a-zA-Z0-9_]+[a-zA-Z0-9]$"
              }
              case "anyUppercaseLetter" => {
                //所有大写字母
                pattern = "^[A-Z]+$"
              }
              case "anyLowercaseLetter" => {
                //所有小写字母
                pattern = "^[a-z]+$"
              }
              case "anyNumberOrLetter" => {
                //所有数字和字母
                pattern = "^[A-Za-z0-9]+$"
              }
              case "dateYYYYMMDD" => {
                //日期格式YYYYMMDD
                pattern = "^\\d{4}\\d{1,2}\\d{1,2}"
              }
              case "dateYYYY-MM-DD" => {
                //日期格式YYYY-MM-DD
                pattern = "^\\d{4}-\\d{1,2}-\\d{1,2}"
              }
              case "dateYYYY/MM/DD" => {
                //日期格式YYYY/MM/DD
                pattern = "^\\d{4}/\\d{1,2}/\\d{1,2}"
              }
              case "dateYY/MM/DD" => {
                //日期格式YY/MM/DD
                pattern = "^\\d{2}/\\d{1,2}/\\d{1,2}"
              }
              case "dateYY-MM-DD" => {
                //日期格式YY-MM-DD
                pattern = "^\\d{2}-\\d{1,2}-\\d{1,2}"
              }
              case "phoneNumber" => {
                //匹配电话号码
                pattern = "^(13[0-9]|14[5|7]|15[0|1|2|3|5|6|7|8|9]|18[0|1|2|3|5|6|7|8|9])\\d{8}$"
              }
              case "email" => {
                //匹配电子邮件
                pattern = "^\\w+([-+.]\\w+)*@\\w+([-.]\\w+)*\\.\\w+([-.]\\w+)*$"

              }
              case "IDCard" => {
                //匹配身份证号
                pattern = "(^\\d{15}$)|(^\\d{18}$)|(^\\d{17}(\\d|X|x)$)"
              }
            }
            val r = pattern.r
            if (!r.pattern.matcher(matchValue).matches()) {
              records += 1
            }
          }
        })
        if (condition equals ("and")) {
          if (records > 0) {
            rows += line
          }
        } else {
          if (records == expressions.length) {
            rows += line
          }
        }
        rows
      })(encoder)
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
      }
    }
    errorData
  }
}

object DataFormatCheck {
  def main(args: Array[String]): Unit = {
    val strings = "3"
    if (!strings.equals("#")) {
      println(strings.toInt)
    }
  }
}
